package com.itbaizhan.hdfs2hbase;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;


import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;


import java.io.IOException;

/**
 * @Auther: hwh
 * @Date: 2023-02-28 - 02 - 28 - 10:41
 * @Description: com.itbaizhan.hdfs2hbase
 * @version: 1.0
 */
public class Hdfs2HbaseMain {
    //这是一个main方法，是程序的入口：
    public static void main(String[] args) throws Exception {
        //加载默认的配置
        Configuration conf = new Configuration(true);
        //设置本地运行
        conf.set("mapreduce.framework.name", "local");
        //指定hbase的zk集群
        conf.set("hbase.zookeeper.quorum", "node02,node03,node04");
        //创建job对象
        Job job = Job.getInstance(conf, "hdfs2hbase demo");
        //指定入口类
        job.setJarByClass(Hdfs2HbaseMain.class);
        //指定输入文件路径 在resource里已经导入了xml文件，默认hdfs路径
        FileInputFormat.addInputPath(job, new Path("/usr/" +
                "local/hello.txt"));
        //指定输入文件路径
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);
        job.setMapperClass(Hdfs2HbaseMapper.class);
        //指定Reducer类，以及处理后的数据放入到Hbase的那种表中
        TableMapReduceUtil.initTableReducerJob(
                "wordcount",//表名称
                Hdfs2HbaseReducer.class,//Reducer类
                job,//指定作业的job对象
                null,null,null,null,
                false//false表示不需要将依赖的jar上传到集群
        );
        //提交作业
        job.waitForCompletion(true);

    }
}
